package user.service;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import user.contstant.Constants;
import user.vo.PassTemplate;

/**
 * 消费kafka中的PassTemplate
 */
@Slf4j
@Component
public class ConsumePassTemplate {

    /** pass 相关的 HBase 服务*/
    private final IHBasePassService iHbasePassService;

    @Autowired
    public ConsumePassTemplate(IHBasePassService iHbasePassService) {
        this.iHbasePassService = iHbasePassService;
    }

    //消费者，注入参数
    @KafkaListener(topics = {Constants.TEMPLATE_TOPIC})
    public void receive(@Payload String passTemplate,
                        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){

        log.info("Consumer Receive PassTemplate:{}",passTemplate);

        PassTemplate pt;
        try {
            //将字符串转换为passTemplate对象,反序列化
            pt = JSON.parseObject(passTemplate,PassTemplate.class);
        }catch (Exception e){
            log.error("Parse PassTemplate Error:{}",e.getMessage());
            return;
        }
        log.info("DropPassTemplateToHBase:{}",iHbasePassService.dropPassTemplateToHBase(pt));
    }
}
